#include "stdafx.h"
#include "net/SocketImpl.hpp"
#include "net/StreamSocketImpl.hpp"
#include "stdlib/timestamp.hpp"
using JHCPP::stdlib::CTimeStamp;
#include "utils/numberformatter.hpp"
using JHCPP::utils::CNumberFormatter;

NET_NAMESPACE_BEGIN

CSocketImpl::CSocketImpl() : m_sockfd(JH_INVALID_SOCKET), m_blocking(true)
{
}

CSocketImpl::CSocketImpl(JH_SOCKET sockfd) : m_sockfd(sockfd), m_blocking(true)
{
}

CSocketImpl::~CSocketImpl()
{
	close();
}

CSocketImpl* CSocketImpl::acceptConnection(CSocketAddress& clientAddr)
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	char buffer[CSocketAddress::MAX_ADDRESS_LENGTH];
	struct sockaddr* pSA = reinterpret_cast<struct sockaddr*>(buffer);
	JH_SOCKETLEN saLen = sizeof(buffer);
	JH_SOCKET sd;
	do
	{
		sd = ::accept(m_sockfd, pSA, &saLen);
	} while (sd == JH_INVALID_SOCKET && lastError() == JH_EINTR);

	if (sd != JH_INVALID_SOCKET)
	{
		clientAddr = CSocketAddress(pSA, saLen);
		return new CStreamSocketImpl(sd);
	}
	error(); // throw
	return 0;
}

void CSocketImpl::connect(const CSocketAddress& address)
{
	if (m_sockfd == JH_INVALID_SOCKET)
	{
		init(address.af());
	}

	int rc;
	do
	{
		rc = ::connect(m_sockfd, address.addr(), address.length());
	}while (rc != 0 && lastError() == JH_EINTR);

	if (rc != 0) 
	{
		int err = lastError();
		error(err, address.toString());
	}
}

void CSocketImpl::connect(const CSocketAddress& address, const CTimeSpan& timeout)
{
	if (m_sockfd == JH_INVALID_SOCKET)
	{
		init(address.af());
	}

	setBlocking(false);
	try
	{
		int rc = ::connect(m_sockfd, address.addr(), address.length());
		if (rc != 0)
		{
			int err = lastError();
			if (err != JH_EINPROGRESS && err != JH_EWOULDBLOCK)
				error(err, address.toString());

			if (!poll(timeout, SELECT_READ | SELECT_WRITE | SELECT_ERROR))
				throw TimeoutException("connect timed out", address.toString());

			err = socketError();
			if (err != 0) 
				error(err);
		}
	}
	catch (CException&)
	{
		setBlocking(true);
		throw;
	}
	setBlocking(true);
}

void CSocketImpl::connectNB(const CSocketAddress& address)
{
	if (m_sockfd == JH_INVALID_SOCKET)
	{
		init(address.af());
	}

	setBlocking(false);
	int rc = ::connect(m_sockfd, address.addr(), address.length());

	if (rc != 0)
	{
		int err = lastError();
		if (err != JH_EINPROGRESS && err != JH_EWOULDBLOCK)
			error(err, address.toString());
	}
}

void CSocketImpl::bind(const CSocketAddress& address, bool reuseAddress)
{
	if (m_sockfd == JH_INVALID_SOCKET)
	{
		init(address.af());
	}

	if (reuseAddress)
	{
		setReuseAddress(true);
		setReusePort(true);
	}

	int rc = ::bind(m_sockfd, address.addr(), address.length());
	if (rc != 0) 
		error(address.toString());
}

void CSocketImpl::bind6(const CSocketAddress& address, bool reuseAddress, bool ipV6Only)
{
#if defined(JH_HAVE_IPv6)
	if (address.family() != CIPAddress::IPv6)
		throw InvalidArgumentException("SocketAddress must be an IPv6 address");

	if (m_sockfd == JH_INVALID_SOCKET)
	{
		init(address.af());
	}

#ifdef IPV6_V6ONLY
	setOption(IPPROTO_IPV6, IPV6_V6ONLY, ipV6Only ? 1 : 0);
#else
	if (ipV6Only) 
		throw NotImplementedException("IPV6_V6ONLY not defined.");
#endif
	if (reuseAddress)
	{
		setReuseAddress(true);
		setReusePort(true);
	}
	int rc = ::bind(m_sockfd, address.addr(), address.length());
	if (rc != 0) error(address.toString());
#else
	throw NotImplementedException("No IPv6 support available");
#endif
}

void CSocketImpl::listen(int backlog)
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	int rc = ::listen(m_sockfd, backlog);
	if (rc != 0) 
		error();
}

void CSocketImpl::close()
{
	if (m_sockfd != JH_INVALID_SOCKET)
	{
		JH_CLOSESOCKET(m_sockfd);
		m_sockfd = JH_INVALID_SOCKET;
	}
}

void CSocketImpl::shutdownReceive()
{
	if (m_sockfd == JH_INVALID_SOCKET)
		throw InvalidSocketException();

	int rc = ::shutdown(m_sockfd, 0);
	if (rc != 0) 
		error();
}


void CSocketImpl::shutdownSend()
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	int rc = ::shutdown(m_sockfd, 1);
	if (rc != 0) 
		error();
}

void CSocketImpl::shutdown()
{
	if (m_sockfd == JH_INVALID_SOCKET)
		throw InvalidSocketException();

	int rc = ::shutdown(m_sockfd, 2);
	if (rc != 0)
		error();
}

int CSocketImpl::sendBytes(const void* buffer, int length, int flags)
{
	if (m_sndTimeout.totalMicroseconds() != 0)
	{
		if (!poll(m_sndTimeout, SELECT_WRITE))
			throw TimeoutException();
	}

	int rc;
	do
	{
		if (m_sockfd == JH_INVALID_SOCKET)
			throw InvalidSocketException();

		rc = ::send(m_sockfd, reinterpret_cast<const char*>(buffer), length, flags);
	}while (m_blocking && rc < 0 && lastError() == JH_EINTR);

	if (rc < 0) 
		error();

	return rc;
}

int CSocketImpl::receiveBytes(void* buffer, int length, int flags)
{
	if (m_recvTimeout.totalMicroseconds() != 0)
	{
		if (!poll(m_recvTimeout, SELECT_READ))
			throw TimeoutException();
	}

	int rc;
	do
	{
		if (m_sockfd == JH_INVALID_SOCKET) 
			throw InvalidSocketException();

		rc = ::recv(m_sockfd, reinterpret_cast<char*>(buffer), length, flags);
	} while (m_blocking && rc < 0 && lastError() == JH_EINTR);

	if (rc < 0) 
	{
		int err = lastError();
		if (err == JH_EAGAIN && !m_blocking)
			;
		else if (err == JH_EAGAIN || err == JH_ETIMEDOUT)
			throw TimeoutException(err);
		else
			error(err);
	}
	return rc;
}

int CSocketImpl::sendTo(const void* buffer, int length, const CSocketAddress& address, int flags)
{
	int rc;
	do
	{
		if (m_sockfd == JH_INVALID_SOCKET) 
			throw InvalidSocketException();

		rc = ::sendto(m_sockfd, reinterpret_cast<const char*>(buffer), length, flags, address.addr(), address.length());
	} while (m_blocking && rc < 0 && lastError() == JH_EINTR);

	if (rc < 0) 
		error();

	return rc;
}

int CSocketImpl::receiveFrom(void* buffer, int length, CSocketAddress& address, int flags)
{
	if (m_recvTimeout.totalMicroseconds() != 0)
	{
		if (!poll(m_recvTimeout, SELECT_READ))
			throw TimeoutException();
	}

	char abuffer[CSocketAddress::MAX_ADDRESS_LENGTH];
	struct sockaddr* pSA = reinterpret_cast<struct sockaddr*>(abuffer);
	JH_SOCKETLEN saLen = sizeof(abuffer);
	int rc;
	do
	{
		if (m_sockfd == JH_INVALID_SOCKET) 
			throw InvalidSocketException();

		rc = ::recvfrom(m_sockfd, reinterpret_cast<char*>(buffer), length, flags, pSA, &saLen);
	} while (m_blocking && rc < 0 && lastError() == JH_EINTR);

	if (rc >= 0)
	{
		address = CSocketAddress(pSA, saLen);
	}
	else
	{
		int err = lastError();
		if (err == JH_EAGAIN && !m_blocking)
			;
		else if (err == JH_EAGAIN || err == JH_ETIMEDOUT)
			throw TimeoutException(err);
		else
			error(err);
	}
	return rc;
}

void CSocketImpl::sendUrgent(unsigned char data)
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	int rc = ::send(m_sockfd, reinterpret_cast<const char*>(&data), sizeof(data), MSG_OOB);
	if (rc < 0) 
		error();
}

int CSocketImpl::available()
{
	int result;
	ioctl(FIONREAD, result);
	return result;
}

bool CSocketImpl::secure() const
{
	return false;
}

bool CSocketImpl::poll(const CTimeSpan& timeout, int mode)
{
	JH_SOCKET sockfd = m_sockfd;
	if (sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	fd_set fdRead;
	fd_set fdWrite;
	fd_set fdExcept;
	FD_ZERO(&fdRead);
	FD_ZERO(&fdWrite);
	FD_ZERO(&fdExcept);
	if (mode & SELECT_READ)
	{
		FD_SET(sockfd, &fdRead);
	}
	if (mode & SELECT_WRITE)
	{
		FD_SET(sockfd, &fdWrite);
	}
	if (mode & SELECT_ERROR)
	{
		FD_SET(sockfd, &fdExcept);
	}

	CTimeSpan remainingTime(timeout);
	int errorCode = JH_ENOERR;
	int rc;
	do
	{
		struct timeval tv;
		tv.tv_sec  = (long) remainingTime.totalSeconds();
		tv.tv_usec = (long) remainingTime.useconds();
		CTimeStamp start;
		rc = ::select(int(sockfd) + 1, &fdRead, &fdWrite, &fdExcept, &tv);
		if (rc < 0 && (errorCode = lastError()) == JH_EINTR)
		{
			CTimeStamp end;
			CTimeSpan waited = end - start;
			if (waited < remainingTime)
				remainingTime -= waited;
			else
				remainingTime = 0;
		}
	} while (rc < 0 && errorCode == JH_EINTR);

	if (rc < 0)
		error(errorCode);

	return rc > 0; 
}

void CSocketImpl::setSendBufferSize(int size)
{
	setOption(SOL_SOCKET, SO_SNDBUF, size);
}

int CSocketImpl::getSendBufferSize()
{
	int result;
	getOption(SOL_SOCKET, SO_SNDBUF, result);
	return result;
}

void CSocketImpl::setReceiveBufferSize(int size)
{
	setOption(SOL_SOCKET, SO_RCVBUF, size);
}

int CSocketImpl::getReceiveBufferSize()
{
	int result;
	getOption(SOL_SOCKET, SO_RCVBUF, result);
	return result;
}

void CSocketImpl::setSendTimeout(const CTimeSpan& timeout)
{
	m_sndTimeout = timeout;
	//setOption(SOL_SOCKET, SO_SNDTIMEO, timeout);
}

CTimeSpan CSocketImpl::getSendTimeout()
{
	CTimeSpan result;
	result = m_sndTimeout;
	//getOption(SOL_SOCKET, SO_SNDTIMEO, result);
	return result;
}

void CSocketImpl::setReceiveTimeout(const CTimeSpan& timeout)
{
	//setOption(SOL_SOCKET, SO_RCVTIMEO, timeout);
	m_recvTimeout = timeout;
}

CTimeSpan CSocketImpl::getReceiveTimeout()
{
	CTimeSpan result;
	result = m_recvTimeout;
	//getOption(SOL_SOCKET, SO_RCVTIMEO, result);
	return result;
}

CSocketAddress CSocketImpl::address()
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	char buffer[CSocketAddress::MAX_ADDRESS_LENGTH];
	struct sockaddr* pSA = reinterpret_cast<struct sockaddr*>(buffer);
	JH_SOCKETLEN saLen = sizeof(buffer);
	int rc = ::getsockname(m_sockfd, pSA, &saLen);
	if (rc == 0)
		return CSocketAddress(pSA, saLen);
	else 
		error();
	return CSocketAddress();
}

CSocketAddress CSocketImpl::peerAddress()
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	char buffer[CSocketAddress::MAX_ADDRESS_LENGTH];
	struct sockaddr* pSA = reinterpret_cast<struct sockaddr*>(buffer);
	JH_SOCKETLEN saLen = sizeof(buffer);
	int rc = ::getpeername(m_sockfd, pSA, &saLen);
	if (rc == 0)
		return CSocketAddress(pSA, saLen);
	else 
		error();
	return CSocketAddress();
}

void CSocketImpl::setOption(int level, int option, int value)
{
	setRawOption(level, option, &value, sizeof(value));
}

void CSocketImpl::setOption(int level, int option, unsigned value)
{
	setRawOption(level, option, &value, sizeof(value));
}

void CSocketImpl::setOption(int level, int option, unsigned char value)
{
	setRawOption(level, option, &value, sizeof(value));
}


void CSocketImpl::setOption(int level, int option, const CIPAddress& value)
{
	setRawOption(level, option, value.addr(), value.length());
}

void CSocketImpl::setOption(int level, int option, const CTimeSpan& value)
{
	struct timeval tv;
	tv.tv_sec  = (long) value.totalSeconds();
	tv.tv_usec = (long) value.useconds();

	setRawOption(level, option, &tv, sizeof(tv));
}

void CSocketImpl::setRawOption(int level, int option, const void* value, JH_SOCKETLEN length)
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();
	
	int rc = ::setsockopt(m_sockfd, level, option, reinterpret_cast<const char*>(value), length);
	if (rc == -1) 
		error();
}

void CSocketImpl::getOption(int level, int option, int& value)
{
	JH_SOCKETLEN len = sizeof(value);
	getRawOption(level, option, &value, len);
}

void CSocketImpl::getOption(int level, int option, unsigned& value)
{
	JH_SOCKETLEN len = sizeof(value);
	getRawOption(level, option, &value, len);
}

void CSocketImpl::getOption(int level, int option, unsigned char& value)
{
	JH_SOCKETLEN len = sizeof(value);
	getRawOption(level, option, &value, len);
}

void CSocketImpl::getOption(int level, int option, CTimeSpan& value)
{
	struct timeval tv;
	JH_SOCKETLEN len = sizeof(tv);
	getRawOption(level, option, &tv, len);
	value.assign(tv.tv_sec, tv.tv_usec);
}

void CSocketImpl::getOption(int level, int option, CIPAddress& value)
{
	char buffer[CIPAddress::MAX_ADDRESS_LENGTH];
	JH_SOCKETLEN len = sizeof(buffer);
	getRawOption(level, option, buffer, len);
	value = CIPAddress(buffer, len);
}

void CSocketImpl::getRawOption(int level, int option, void* value, JH_SOCKETLEN& length)
{
	if (m_sockfd == JH_INVALID_SOCKET) 
		throw InvalidSocketException();

	int rc = ::getsockopt(m_sockfd, level, option, reinterpret_cast<char*>(value), &length);
	if (rc == -1) 
		error();
}

void CSocketImpl::setLinger(bool on, int seconds)
{
	struct linger l;
	l.l_onoff  = on ? 1 : 0;
	l.l_linger = seconds;
	setRawOption(SOL_SOCKET, SO_LINGER, &l, sizeof(l));
}

void CSocketImpl::getLinger(bool& on, int& seconds)
{
	struct linger l;
	JH_SOCKETLEN len = sizeof(l);
	getRawOption(SOL_SOCKET, SO_LINGER, &l, len);
	on      = l.l_onoff != 0;
	seconds = l.l_linger;
}

void CSocketImpl::setNoDelay(bool flag)
{
	int value = flag ? 1 : 0;
	setOption(IPPROTO_TCP, TCP_NODELAY, value);
}

bool CSocketImpl::getNoDelay()
{
	int value(0);
	getOption(IPPROTO_TCP, TCP_NODELAY, value);
	return value != 0;
}

void CSocketImpl::setKeepAlive(bool flag)
{
	int value = flag ? 1 : 0;
	setOption(SOL_SOCKET, SO_KEEPALIVE, value);
}

bool CSocketImpl::getKeepAlive()
{
	int value(0);
	getOption(SOL_SOCKET, SO_KEEPALIVE, value);
	return value != 0;
}

void CSocketImpl::setReuseAddress(bool flag)
{
	int value = flag ? 1 : 0;
	setOption(SOL_SOCKET, SO_REUSEADDR, value);
}

bool CSocketImpl::getReuseAddress()
{
	int value(0);
	getOption(SOL_SOCKET, SO_REUSEADDR, value);
	return value != 0;
}

void CSocketImpl::setReusePort(bool flag)
{
#ifdef SO_REUSEPORT
	try
	{
		int value = flag ? 1 : 0;
		setOption(SOL_SOCKET, SO_REUSEPORT, value);
	}
	catch (IOException&)
	{
	}
#endif
}

bool CSocketImpl::getReusePort()
{
#ifdef SO_REUSEPORT
	int value(0);
	getOption(SOL_SOCKET, SO_REUSEPORT, value);
	return value != 0;
#else
	return false;
#endif
}

void CSocketImpl::setOOBInline(bool flag)
{
	int value = flag ? 1 : 0;
	setOption(SOL_SOCKET, SO_OOBINLINE, value);
}

bool CSocketImpl::getOOBInline()
{
	int value(0);
	getOption(SOL_SOCKET, SO_OOBINLINE, value);
	return value != 0;
}

void CSocketImpl::setBroadcast(bool flag)
{
	int value = flag ? 1 : 0;
	setOption(SOL_SOCKET, SO_BROADCAST, value);
}


bool CSocketImpl::getBroadcast()
{
	int value(0);
	getOption(SOL_SOCKET, SO_BROADCAST, value);
	return value != 0;
}

void CSocketImpl::setBlocking(bool flag)
{
#if !defined(JH_OS_LINUX)
	int arg = flag ? 0 : 1;
	ioctl(FIONBIO, arg);
#else
	int arg = fcntl(F_GETFL);
	long flags = arg & ~O_NONBLOCK;
	if (!flag) flags |= O_NONBLOCK;
	(void) fcntl(F_SETFL, flags);
#endif
	m_blocking = flag;
}

bool CSocketImpl::getBlocking() const
{
	return m_blocking;
}

int CSocketImpl::socketError()
{
	int result(0);
	getOption(SOL_SOCKET, SO_ERROR, result);
	return result;
}

JH_SOCKET CSocketImpl::sockfd() const
{
	return m_sockfd;
}

bool CSocketImpl::initialized() const
{
	return m_sockfd != JH_INVALID_SOCKET;
}

void CSocketImpl::init(int af)
{
	initSocket(af, SOCK_STREAM);
}

void CSocketImpl::initSocket(int af, int type, int proto)
{
	jh_assert(m_sockfd == JH_INVALID_SOCKET);

	m_sockfd = ::socket(af, type, proto);
	if (m_sockfd == JH_INVALID_SOCKET)
		error();
}

void CSocketImpl::ioctl(int request, int& arg)
{
#if defined(_WIN32)
	int rc = ioctlsocket(m_sockfd, request, reinterpret_cast<u_long*>(&arg));
#else
	int rc = ::ioctl(m_sockfd, request, &arg);
#endif
	if (rc != 0) 
		error();
}

void CSocketImpl::ioctl(int request, void* arg)
{
#if defined(_WIN32)
	int rc = ioctlsocket(m_sockfd, request, reinterpret_cast<u_long*>(arg));
#else
	int rc = ::ioctl(m_sockfd, request, arg);
#endif
	if (rc != 0) 
		error();
}

#if defined(JH_OS_LINUX)
int CSocketImpl::fcntl(int request)
{
	int rc = ::fcntl(m_sockfd, request);
	if (rc == -1) 
		error();
	return rc;
}

int CSocketImpl::fcntl(int request, long arg)
{
	int rc = ::fcntl(m_sockfd, request, arg);
	if (rc == -1)
		error();
	return rc;
}
#endif


void CSocketImpl::reset(JH_SOCKET aSocket)
{
	m_sockfd = aSocket;
}

int CSocketImpl::lastError()
{
#if defined(_WIN32)
	return WSAGetLastError();
#else
	return errno;
#endif
}

void CSocketImpl::error()
{
	int err = lastError();
	std::string empty;
	error(err, empty);
}

void CSocketImpl::error(const std::string& arg)
{
	error(lastError(), arg);
}

void CSocketImpl::error(int code)
{
	std::string arg;
	error(code, arg);
}

void CSocketImpl::error(int code, const std::string& arg)
{
	switch (code)
	{
	case JH_ENOERR: return;
	case JH_ESYSNOTREADY:
		throw NetException("Net subsystem not ready", code);
	case JH_ENOTINIT:
		throw NetException("Net subsystem not initialized", code);
	case JH_EINTR:
		throw IOException("Interrupted", code);
	case JH_EACCES:
		throw IOException("Permission denied", code);
	case JH_EFAULT:
		throw IOException("Bad address", code);
	case JH_EINVAL:
		throw InvalidArgumentException(code);
	case JH_EMFILE:
		throw IOException("Too many open files", code);
	case JH_EWOULDBLOCK:
		throw IOException("Operation would block", code);
	case JH_EINPROGRESS:
		throw IOException("Operation now in progress", code);
	case JH_EALREADY:
		throw IOException("Operation already in progress", code);
	case JH_ENOTSOCK:
		throw IOException("Socket operation attempted on non-socket", code);
	case JH_EDESTADDRREQ:
		throw NetException("Destination address required", code);
	case JH_EMSGSIZE:
		throw NetException("Message too long", code);
	case JH_EPROTOTYPE:
		throw NetException("Wrong protocol type", code);
	case JH_ENOPROTOOPT:
		throw NetException("Protocol not available", code);
	case JH_EPROTONOSUPPORT:
		throw NetException("Protocol not supported", code);
	case JH_ESOCKTNOSUPPORT:
		throw NetException("Socket type not supported", code);
	case JH_ENOTSUP:
		throw NetException("Operation not supported", code);
	case JH_EPFNOSUPPORT:
		throw NetException("Protocol family not supported", code);
	case JH_EAFNOSUPPORT:
		throw NetException("Address family not supported", code);
	case JH_EADDRINUSE:
		throw NetException("Address already in use", arg, code);
	case JH_EADDRNOTAVAIL:
		throw NetException("Cannot assign requested address", arg, code);
	case JH_ENETDOWN:
		throw NetException("Network is down", code);
	case JH_ENETUNREACH:
		throw NetException("Network is unreachable", code);
	case JH_ENETRESET:
		throw NetException("Network dropped connection on reset", code);
	case JH_ECONNABORTED:
		throw ConnectionAbortedException(code);
	case JH_ECONNRESET:
		throw ConnectionResetException(code);
	case JH_ENOBUFS:
		throw IOException("No buffer space available", code);
	case JH_EISCONN:
		throw NetException("Socket is already connected", code);
	case JH_ENOTCONN:
		throw NetException("Socket is not connected", code);
	case JH_ESHUTDOWN:
		throw NetException("Cannot send after socket shutdown", code);
	case JH_ETIMEDOUT:
		throw TimeoutException(code);
	case JH_ECONNREFUSED:
		throw ConnectionRefusedException(arg, code);
	case JH_EHOSTDOWN:
		throw NetException("Host is down", arg, code);
	case JH_EHOSTUNREACH:
		throw NetException("No route to host", arg, code);
#if defined(JH_OS_LINUX)
	case EPIPE:
		throw IOException("Broken pipe", code);
	case EBADF:
		throw IOException("Bad socket descriptor", code);
#endif
	default:
		throw IOException(CNumberFormatter::format(code), arg, code);
	}
}

NET_NAMESPACE_END
